﻿using System;
using System.Collections.Generic;
using System.Text;
using Apache.Cassandra;
using Thrift.Protocol;
using Thrift.Transport;

namespace Bbm.Persistence.Cassandra
{
	public class CassandraStoreFactory : IPhysicalStoreFactory
	{
		readonly string keySpace;
		readonly string columnFamily;
		readonly Apache.Cassandra.Cassandra.Client client;

		public CassandraStoreFactory(string host, int port, string keySpace, string columnFamily)
		{
			this.keySpace = keySpace;
			this.columnFamily = columnFamily;
			var transport = new TSocket(host, port);
			var protocol = new TBinaryProtocol(transport);

			client = new Apache.Cassandra.Cassandra.Client(protocol);
			transport.Open();
		}

		/// <summary>
		///  Returns a physical store for the given storage name. Name uniqueness
		///  is provided by the caller.
		///  </summary>
		/// <param name="name">
		///   The name of the store
		/// </param>
		/// <returns>
		///  A PhysicalStoreBase object the allows persistent storage
		///  </returns>
		public IDictionary<string, string> GetStoreForName(CacheStoreName name)
		{
			return new CassandraStore(client, keySpace,
			                          new ColumnPath
			                          	{
			                          		Column_family = columnFamily,
			                          		Column = Encoding.UTF8.GetBytes(name.ToString())
			                          	});
		}
	}

	class CassandraStore : PhysicalStoreBase
	{
		readonly Apache.Cassandra.Cassandra.Client client;
		readonly string keySpace;
		readonly ColumnPath path;

		public CassandraStore(Apache.Cassandra.Cassandra.Client client, string keySpace, ColumnPath path)
		{
			this.client = client;
			this.keySpace = keySpace;
			this.path = path;
		}

		/// <summary>
		/// Determines whether the <see cref="T:System.Collections.Generic.IDictionary`2"/> contains an element with the specified key.
		/// </summary>
		/// <returns>
		/// true if the <see cref="T:System.Collections.Generic.IDictionary`2"/> contains an element with the key; otherwise, false.
		/// </returns>
		/// <param name="key">The key to locate in the <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		///                 </param><exception cref="T:System.ArgumentNullException"><paramref name="key"/> is null.
		///                 </exception>
		public override bool ContainsKey(string key)
		{
			if (key == null) 
				throw new ArgumentNullException("key");

			try
			{
				client.get(keySpace, key, path, ConsistencyLevel.ONE);
			}
			catch (NotFoundException)
			{
				return false;
			}

			return true;
		}

		/// <summary>
		/// Adds an element with the provided key and value to the <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		/// </summary>
		/// <param name="key">The object to use as the key of the element to add.
		///                 </param><param name="value">The object to use as the value of the element to add.
		///                 </param><exception cref="T:System.ArgumentNullException"><paramref name="key"/> is null.
		///                 </exception><exception cref="T:System.ArgumentException">An element with the same key already exists in the <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		///                 </exception><exception cref="T:System.NotSupportedException">The <see cref="T:System.Collections.Generic.IDictionary`2"/> is read-only.
		///                 </exception>
		public override void Add(string key, string value)
		{
			if (key == null) 
				throw new ArgumentNullException("key");

			if(ContainsKey(key))
				throw new ArgumentException("An element with the same key already exists in the collection", "key");

			client.insert(keySpace, key, path, Encoding.UTF8.GetBytes(value), DateTime.Now.ToFileTime(), ConsistencyLevel.ONE);
		}

		/// <summary>
		/// Removes the element with the specified key from the <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		/// </summary>
		/// <returns>
		/// true if the element is successfully removed; otherwise, false.  This method also returns false if <paramref name="key"/> was not found in the original <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		/// </returns>
		/// <param name="key">The key of the element to remove.
		///                 </param><exception cref="T:System.ArgumentNullException"><paramref name="key"/> is null.
		///                 </exception><exception cref="T:System.NotSupportedException">The <see cref="T:System.Collections.Generic.IDictionary`2"/> is read-only.
		///                 </exception>
		public override bool Remove(string key)
		{
			if (key == null) 
				throw new ArgumentNullException("key");

			if(!ContainsKey(key))
				return false;

			client.remove(keySpace, key, path, DateTime.Now.ToFileTime(), ConsistencyLevel.ONE);

			return true;
		}

		/// <summary>
		/// Gets or sets the element with the specified key.
		/// </summary>
		/// <returns>
		/// The element with the specified key.
		/// </returns>
		/// <param name="key">The key of the element to get or set.
		///                 </param><exception cref="T:System.ArgumentNullException"><paramref name="key"/> is null.
		///                 </exception><exception cref="T:System.Collections.Generic.KeyNotFoundException">The property is retrieved and <paramref name="key"/> is not found.
		///                 </exception><exception cref="T:System.NotSupportedException">The property is set and the <see cref="T:System.Collections.Generic.IDictionary`2"/> is read-only.
		///                 </exception>
		public override string this[string key]
		{
			get
			{
				if (key == null) 
					throw new ArgumentNullException("key");

				try
				{
					return Encoding.UTF8.GetString(client.get(keySpace, key, path, ConsistencyLevel.ONE).Column.Value);
				}
				catch(NotFoundException)
				{
					throw new KeyNotFoundException();
				}
			}

			set
			{
				if (key == null) 
					throw new ArgumentNullException("key");

				client.insert(keySpace, key, path, Encoding.UTF8.GetBytes(value), DateTime.Now.ToFileTime(), ConsistencyLevel.ONE);
			}
		}

		/// <summary>
		/// Gets an <see cref="T:System.Collections.Generic.ICollection`1"/> containing the keys of the <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		/// </summary>
		/// <returns>
		/// An <see cref="T:System.Collections.Generic.ICollection`1"/> containing the keys of the object that implements <see cref="T:System.Collections.Generic.IDictionary`2"/>.
		/// </returns>
		public override ICollection<string> Keys
		{
			get { throw new NotImplementedException(); }
		}
	}
}
